package groupManagementModule;

import generalObjects.CassandraHandler;
import generalObjects.GroupView;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import messageOrderingModule.CausalAlgorithm;

import org.codehaus.jackson.type.JavaType;

import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import communicationModule.MessageObject;
import communicationModule.SendResult;

public class HistoryRetriever {

    private final CassandraHandler dbCom;
    private final String nickname;
    private final String groupname;
    private final ConcurrentHashMap<Long, Integer> receivedbuf;
    private ConcurrentHashMap<String, Integer> vclkGroup;
    private final List<SendResult> history = new ArrayList<SendResult>();
    private final JavaType type;

    private final String getLastSeenClock = "SELECT lastseenvclock FROM "
            + "usersstamp WHERE nick=? AND groupname=? ALLOW FILTERING";
    private final String getLastGroupClock = "SELECT myseqnr,nick FROM "
            + "usersstamp WHERE groupname=? ALLOW FILTERING";
    private final String getMsgs = "SELECT * FROM grouphistory "
            + "WHERE senderseqnr>=? AND groupname=? AND sender=? ALLOW FILTERING";

    public HistoryRetriever(final CassandraHandler dbCom,
            final String nickname, final String groupname,
            final ConcurrentHashMap<Long, Integer> receivedbuf) {
        this.dbCom = dbCom;
        this.nickname = nickname;
        this.groupname = groupname;
        this.receivedbuf = receivedbuf;
        type = dbCom
                .getObjectMapper()
                .getTypeFactory()
                .constructMapType(ConcurrentHashMap.class, String.class,
                        Integer.class);
    }

    @SuppressWarnings("unchecked")
    private ConcurrentHashMap<String, Integer> retrieveLastClock() {

        final ResultSet vr = dbCom.executeStatement(getLastSeenClock,
                ConsistencyLevel.QUORUM, nickname, groupname);
        // if there is not stamp for this user saved.
        if (vr.isExhausted()) {
            return null;
        }
        final String vclkJSON = vr.one().getString("lastseenvclock");
        return (ConcurrentHashMap<String, Integer>) dbCom.fromJSONString(
                vclkJSON, type);
    }

    public ConcurrentHashMap<String, Integer> getGroupClock() {
        return vclkGroup;
    }

    private void retrieveLastGroupClock() {

        vclkGroup = new ConcurrentHashMap<String, Integer>();

        final ResultSet clock = dbCom.executeStatement(getLastGroupClock,
                ConsistencyLevel.QUORUM, groupname);
        for (final Row row : clock) {
            final int membersSeqNr = row.getInt("myseqnr");
            final String memberNick = row.getString("nick");
            vclkGroup.put(memberNick, membersSeqNr);
        }

    }

    public List<SendResult> retrieveMsgs() {

        final ConcurrentHashMap<String, Integer> vclkstamp = retrieveLastClock();
        retrieveLastGroupClock();
        // new user == no history retrieval.
        if (vclkstamp == null) {
            return null;
        }

        for (final Map.Entry<String, Integer> entry : vclkGroup.entrySet()) {

            final String nickofMember = entry.getKey();
            if (vclkstamp.get(nickofMember) == null) {
                // get all msgs from the new user. meaning sequence nr >= 0
                vclkstamp.put(nickofMember, 0);
                final ResultSet msgs = dbCom.executeStatement(getMsgs,
                        ConsistencyLevel.QUORUM, 1, groupname, nickofMember);
                buildMsgs(msgs);

            } else {

                final ResultSet msgs = dbCom.executeStatement(getMsgs,
                        ConsistencyLevel.QUORUM,
                        (vclkstamp.get(entry.getKey()) + 1), groupname,
                        nickofMember);
                buildMsgs(msgs);

            }

        }
        return orderCausally(vclkstamp);

    }

    private List<SendResult> orderCausally(
            final ConcurrentHashMap<String, Integer> vclkstamp) {
        final GroupView gV = new GroupView();
        gV.setVectorClock(vclkstamp);
        final CausalAlgorithm ca = new CausalAlgorithm(gV, new AtomicBoolean());
        List<SendResult> histmsgs = ca.getDeliverables(history);
        // stamp seen history
        final String insertStamp = "INSERT INTO usersstamp "
                + "(groupname,nick,myseqnr,lastseenvclock) VALUES(?,?,?,?)";
        int myseqnr = gV.getVectorClock().get(nickname);
        final String stampclockJSON = dbCom
                .mapToJSONString(gV.getVectorClock());
        dbCom.executeStatement(insertStamp, ConsistencyLevel.QUORUM, groupname,
                nickname, myseqnr, stampclockJSON);

        return histmsgs;

    }

    private void buildMsgs(final ResultSet msgs) {
        for (final Row row : msgs) {
            final Long id = row.getLong("id");
            final String header = row.getString("header");
            final String data = row.getString("data");
            final String sender = row.getString("sender");
            final String msgvclockJSON = row.getString("vectorclock");

            @SuppressWarnings("unchecked")
            final ConcurrentHashMap<String, Integer> msgvclock = (ConcurrentHashMap<String, Integer>) dbCom
                    .fromJSONString(msgvclockJSON, type);
            final GroupView msgGView = new GroupView();
            msgGView.setVectorClock(msgvclock);
            // reconstruct msg
            final SendResult msgObj = new SendResult(new MessageObject(sender,
                    header, data, msgGView, ""), null);
            // add to recievebuf so we dont get duplicates(incase they
            // exist in workqueue).
            receivedbuf.put(id, 1);
            history.add(msgObj);
        }
    }
}
